package org.artifact.core.plugin.rocketmq;

import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.client.producer.SendResult;
import org.apache.rocketmq.client.producer.selector.SelectMessageQueueByHash;
import org.apache.rocketmq.common.message.Message;

import java.util.HashMap;
import java.util.Map;

public class RocketMQProducer {
    private String namesrvAddr = null;
    private String groupName = null;
    private DefaultMQProducer producer = null;

    public void start() {
        //声明并初始化一个producer
        //需要一个producer group名字作为构造方法的参数，这里为producer1
        producer = new DefaultMQProducer(groupName);

        //设置NameServer地址,此处应改为实际NameServer地址，多个地址之间用；分隔
        //NameServer的地址必须有，但是也可以通过环境变量的方式设置，不一定非得写死在代码里
        producer.setNamesrvAddr(namesrvAddr);

        //调用start()方法启动一个producer实例
        try {
            producer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }
    }

    public void send(Message msg) {
        try {
//            producer.

            String orderId = msg.getUserProperty("orderId");
            //调用producer的send()方法发送消息
            //这里调用的是同步的方式，所以会有返回结果
            SendResult sendResult = producer.send(msg, new SelectMessageQueueByHash(), orderId);
            //打印返回结果，可以看到消息发送的状态以及一些相关信息
            System.out.println(sendResult);
        } catch (Exception e) {
            e.printStackTrace();
        }
    }

    public void stop() {
        //发送完消息之后，调用shutdown()方法关闭producer
        producer.shutdown();
    }

    public static void main(String[] args) {
        RocketMQProducer producer = new RocketMQProducer();
        producer.namesrvAddr = "localhost:9876";
        producer.groupName = "defaultGroup";
        producer.start();

        Map<Integer,Integer> map = new HashMap<>();
        for (int i = 0;i < 100;i++){
            int order = i % 8;
            int index = map.computeIfAbsent(order,k -> 0) + 1;
            map.put(order,index);
            Message msg = new Message("Demo","1001",(index + "").getBytes());
            msg.putUserProperty("orderId",order+"");
            producer.send(msg);
        }

    }
}
